package net.bwie.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.common.utils.DateTimeUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class AdjustIsNewProcessFunction extends KeyedProcessFunction<String, String, String> {


    //step1 声明状态
    private transient ValueState<String> firstDataState = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        //step2 初始化状态
        firstDataState = getRuntimeContext().getState(
                new ValueStateDescriptor<String>("first-date", String.class)
        );

    }


    @Override
    public void processElement(String value,Context ctx, Collector<String> out) throws Exception {
        //step3 获取状态数据
        String firstDate = firstDataState.value();

        //解析字段值

        JSONObject jsonObject = JSON.parseObject(value);
        String isNewValue = jsonObject.getJSONObject("common").getString("is_new");
        Long tsValue = jsonObject.getLong("ts");
        
        // todo isNew = 1   firstDate=null     firstDate=有值
        if ("1".equals(isNewValue)){

            if(null == firstDate){
                //step4 更新状态数据
                //此时的is_new正确，状态更新为第一次访问时间
               firstDataState.update(DateTimeUtil.convertLongToString(tsValue, DateTimeUtil.DATE_FORMAT));
            }else{
                //状态值 （首次访问日期） 等于数据日期  不做处理
                if(!firstDate.equals(DateTimeUtil.convertLongToString(tsValue, DateTimeUtil.DATE_FORMAT))){
                    //此时is_new属于异常，需要纠正 设置为0
                    jsonObject.getJSONObject("common").put("is_new","0");
                }
            }


        }else{
            //当用户早早登录访问的时候，当时实时应用后来处理，用户第一次访问时，firstDate为null
            if(null != firstDate){
               //设置一个昨天的日期
               firstDataState.update(DateTimeUtil.convertLongToString(tsValue- 86400000,DateTimeUtil.DATE_FORMAT));

            }
        }
        //输出数据
        out.collect(jsonObject.toJSONString());
    }




}
